package com.shanks.framework.mq.producer;

import com.shanks.framework.mq.common.constant.MQConstant;
import com.shanks.framework.mq.domain.MQMessage;
import com.shanks.framework.mq.exception.MQException;
import com.shanks.framework.mq.common.util.JsonUtil;
import com.shanks.framework.mq.common.util.MQUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.UnsupportedEncodingException;
import java.util.Map;

public abstract class AbstractMQProducer implements BaseMQProducer {
    private static final Logger logger = LoggerFactory.getLogger(AbstractMQProducer.class);

    private String producerKey;


    public void setProducerKey(String producerKey) {
        this.producerKey = producerKey;
    }

    @Override
    public String getProducerKey() {
        return this.producerKey;
    }

    @Override
    public void send(String topic, String tags, String key, String body, String source) throws MQException {
        try {
            byte[] bytes = body.getBytes("UTF-8");
            //创建签名
            String sign = MQUtil.createSignature(topic, tags, key, bytes, source);
            this.doSend(topic, tags, key, bytes, source, sign);
        } catch (UnsupportedEncodingException e) {
            logger.error("convert body string to byte[] failed, body:{}", body, e);
            throw new MQException(e);
        }
    }

    @Override
    public void sendAck(String originTopic, String originTag, String originKey, Boolean result, String source) throws MQException {
        String ackMsgTopic = MQConstant.TOPIC.ACK_MSG_TOPIC;

        Map<String, Object> messageData = MQMessage.MQMessageBuilder.create()
                .put(MQConstant.PROPERTY.ORIGIN_TOPIC, originTopic)
                .put(MQConstant.PROPERTY.ORIGIN_TAG, originTag)
                .put(MQConstant.PROPERTY.ORIGIN_KEY, originKey)
                .put(MQConstant.PROPERTY.RESULT, result)
                .put(MQConstant.PROPERTY.SOURCE, source)
                .build();

        //添加原始TOPIC对应的TAG
        String mainTag = MQUtil.getAckMsgTag(originTopic, "*");
        //ACK消息的TAG='原始TOPIC_原始TAG || 原始TOPIC'
        String subTag = MQUtil.getAckMsgTag(originTopic, originTag);
        this.send(ackMsgTopic, mainTag, originKey, JsonUtil.toJson(messageData), source);
        this.send(ackMsgTopic, subTag, originKey, JsonUtil.toJson(messageData), source);
    }


    /**
     * 发送消息
     *
     * @param topic
     * @param tags
     * @param key
     * @param body
     * @param source
     * @param sign
     * @throws MQException
     */
    protected abstract void doSend(String topic, String tags, String key, byte[] body, String source, String sign) throws MQException;
}
